package com.huitone.smspfm.interfacz.mo.deliver;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.interfacz.logger.ExtLogger;
import com.huitone.smspfm.interfacz.mo.SvrNodeClientsManager;
import com.huitone.smspfm.interfacz.core.AppStaticConfigs;
import com.huitone.smspfm.interfacz.dao.entity.SvrNodeDef;
import com.huitone.smspfm.interfacz.model.MoDealHash;
import com.huitone.smspfm.interfacz.model.SmsDeliverForResend;
import com.huitone.smspfm.interfacz.utils.SMSContentUtils;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.socket.client.SmsClient;
import com.huitone.smspfm.socket.protocol.sgip.constant.CommandID;
import com.huitone.smspfm.socket.protocol.sgip.constant.Constant;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Deliver;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;

/**   
* @Title ReportResendManager.java 
* @Package com.huitone.smspfm.interfacz.delrep.report 
* @Description TODO(用一句话描述该文件做什么) 
* @author ybliang   
* @date 2017年4月11日 下午5:15:31 
* @version V1.0   
*/
public class SmsDeliverResendTask extends Thread {
	
	private static final ExtLogger logger = ExtLogger.create(SmsDeliverResendTask.class);
	
	private boolean toSleep = false;
	
	private long sleepTime = 5000;
	
	private int poolSize = (int) Math.ceil(AppStaticConfigs.smspfmMoPoolSize / 2);
	
	private ExecutorService executorService = new ThreadPoolExecutor(poolSize, 
			poolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(poolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			if (!executor.isShutdown()) {
				try {
					executor.getQueue().put(r);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	});
	
	private boolean keepGoing = true;

	@Override
	public void run() {
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script = new DefaultRedisScript<>(AppStaticConfigs.luaGetResendMo, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(AppStaticConfigs.iSMMoResendList);
		keys.add(AppStaticConfigs.pSMCacheHash);
		
		while(!this.isInterrupted() && isKeepGoing()) {
			
			executorService.execute(new Runnable() {
				
				@Override
				public void run() {
					
					List<List<String>> lists = AppStaticConfigs.scriptOpts3.getHashList(script, keys, AppStaticConfigs.moReadPerTime + "");
					
					List<SmsDeliverForResend> entities = new ArrayList<SmsDeliverForResend>();
					for (int i = 0 ; i < lists.size() ; i ++) {
						List<String> list = lists.get(i);
						SmsDeliverForResend entity = SmsDeliverForResend.parse(list);
						entities.add(entity);
					}
					
					if (entities.isEmpty()) {
						toSleep = true;
						return;
					}
					
					toSleep = false;
					
					List<MoDealHash> mdhs = new ArrayList<MoDealHash>();
					
					AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {
						
						@Override
						public void excute(RedisConnection connection) {
					
							for (SmsDeliverForResend sd:entities) {
								
								String cacheKey = sd.getSerialno() + ":" + sd.getPhoneNumber();
								
								if (sd.isWellInformed()) {
									
									SvrNodeDef svrNodeDef = AppStaticConfigs.svrNodeDefMap.get(sd.getSrvNode());
									
									if (svrNodeDef != null) {
										
										if (StringUtils.equals(svrNodeDef.getAcceptMo(), "0") || svrNodeDef.getAcceptMo() == null) { // 是否向业务端推送上行短信, 0 推送 1 不推送 2 强制不推送(由于多次连接不成功导致)
											
											SmsClient svrNodeClient = SvrNodeClientsManager.getInstance().getSvrNodeClient(sd.getSrvNode());
											
											if (svrNodeClient != null) {
												
												String orgSrvNode = sd.getOrgSrvNode();
												
												String srvNode = StringUtils.isEmpty(orgSrvNode) ? sd.getSrvNode() : orgSrvNode;
												
												String sendPort = StringUtils.isEmpty(orgSrvNode) ? sd.getSendPort() : "95598" + sd.getSendPort();
													
												Deliver deliver = Deliver.build(srvNode, sendPort, sd.getPhoneNumber(), sd.getRecvTime(), Constant.MessageCodingE.GBK.getValue(), SMSContentUtils.cut70(sd.getSmcontent()));
												
												Head head = Head.build(deliver.length(), CommandID.SGIP_DELIVER, sd.getLastSerialPart1(), sd.getLastSerialPart2(), sd.getLastSerialPart3());
												
												/// ---
												
												MoDealHash dealHash = new MoDealHash();
												
												dealHash.setiSMCacheKey(cacheKey);
												dealHash.setiRptSnPart1(head.getSerialNoPart1());
												dealHash.setiRptSnPart2(head.getSerialNoPart2());
												dealHash.setiRptSnPart3(head.getSerialNoPart3());
												dealHash.setiSendTimes(sd.getSendTimes() + 1);
												dealHash.setiLastSendTime(new Date());
												
												mdhs.add(dealHash);
												
												//String hkey = AppStaticConfigs.pMMoDealTmpDlvHash + ":" + serialNo;
												
												//byte[] hkeyb = AppStaticConfigs.stringRedisSerializer.serialize(hkey);
												
												//connection.hMSet(hkeyb, dealHash.getMapB());
												//connection.expire(hkeyb, 3600);
												//connection.zAdd(AppStaticConfigs.iSMMoDealSerialNoSetB, 
												//		System.currentTimeMillis() + AppStaticConfigs.moResendInterval, AppStaticConfigs.stringRedisSerializer.serialize(head.getSerialNo()));
												
												/// ---
												
												Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
												map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), map);
												
												Message message = new Message(head, deliver);
												
												logger.delivrd(message.toString());
												
												boolean sendDeliver = svrNodeClient.sendMsg(message);
												
												if (!sendDeliver) {
													logger.dfailed("重新推送上行短信失败:" + sd.toString());
												}
													
											} else {
												
												String desc = "重新推送上行短信时,没有匹配到业务端:" + sd.getSrvNode();
												
												connection.lPush(AppStaticConfigs.iSMMoEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey));
												
												Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
												map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
												map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
												map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey),map);
											
												logger.dfailed(desc + sd.toString());
											}
										} else {
											if (StringUtils.equals(svrNodeDef.getAcceptMo(), "2")) {
												logger.dfailed("重新推送上行短信时,业务端:" + svrNodeDef.getSvrNode() + "当前ACCEPT_MO状态为2,连接不可用,直接写上行短信完成队列:" + sd.toString());
											}
											// 直接写上行完成队列
											AppStaticConfigs.listOpts3.in(AppStaticConfigs.iSMMoEndList, cacheKey);
										}
									} else {
										String desc = "重新推送上行短信时,没有匹配到业务端:" + sd.getSrvNode();
										
										connection.lPush(AppStaticConfigs.iSMMoEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey));
										
										Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
										map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
										map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
										map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
										
										connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey),map);
									
										logger.dfailed(desc + sd.toString());
									}
									
								} else {
									String desc = "重新推送上行短信时,读取的上行短信信息不全:";
									connection.lPush(AppStaticConfigs.iSMMoEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey));
									
									Map<byte[], byte[]> map = new HashMap<byte[], byte[]>();
									map.put(AppCanstant.SmsEntity.MoResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
									map.put(AppCanstant.SmsEntity.MoResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
									map.put(AppCanstant.SmsEntity.MoTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(null)));
									
									connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey),map);
								
									logger.dfailed(desc + sd.toString());
								}
							}
						}
					});
					
					// 保存重发上行短信需要到的信息
					AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

						@Override
						public void excute(RedisConnection connection) {
							
							for (MoDealHash mdh:mdhs) {
								
								String hkey = AppStaticConfigs.pMMoDealTmpDlvHash + ":" + mdh.getSerialNo();
								
								byte[] hkeyb = AppStaticConfigs.stringRedisSerializer.serialize(hkey);
								
								connection.hMSet(hkeyb, mdh.getMapB());
								connection.expire(hkeyb, 86400);
								connection.zAdd(AppStaticConfigs.iSMMoDealSerialNoSetB, 
										System.currentTimeMillis() + AppStaticConfigs.moResendInterval, AppStaticConfigs.stringRedisSerializer.serialize(mdh.getSerialNo()));
							}
						}
					});
				}
			});
			
			try {
				if (toSleep) {
					Thread.sleep(sleepTime);
				}
			} catch (InterruptedException e) {
				break;
			}
		}
		
		if (executorService != null && !executorService.isShutdown()) {
			executorService.shutdown();
		}
		
		logger.info("上行短信超时重发任务退出!");
	}
	
	/**
	* @Title: ReportResendManager.java 
	* @Package com.huitone.smspfm.interfacz.delrep.report 
	* @Description: 由于程序Crash或者其它原因,造成在重发过程中可能会出现的SMMoDealSerialNoSet集合序列被删除,但是消息却没有发送成功的情况,这时应该根据SMMoDealTmpDlvHashPrefix哈希结构的数据修复SMMoDealSerialNoSet.
	* @author ybliang   
	* @date 2017年5月31日 下午5:15:12 
	* @version V1.0
	 */
	class ResendSetWatchDog extends Thread {
		
		private long maxInteval = 600000;

		@Override
		public void run() {
			
			while(!this.isInterrupted()) {
				
				Set<String> findKeys = AppStaticConfigs.keyOpts.findKeys(AppStaticConfigs.pMMoDealTmpDlvHash+":*");
				
				for (String key:findKeys) {
					
					MoDealHash moDealHash = MoDealHash.parse(AppStaticConfigs.hashOpts.get(key, MoDealHash.getColumns()));
					
					if (moDealHash.isExists()) {
						
						String serialNo = moDealHash.getSerialNo();
						Date iLastSendTime = moDealHash.getiLastSendTime();
						
						if (iLastSendTime != null && !AppStaticConfigs.zsetOpts.isMember(AppStaticConfigs.iSMMoDealSerialNoSet, serialNo)) {
							long lastTime = iLastSendTime.getTime();
							long currTime = System.currentTimeMillis();
							long inteval = currTime - lastTime;
							if (inteval > maxInteval) {
								AppStaticConfigs.zsetOpts.add(AppStaticConfigs.iSMMoDealSerialNoSet, serialNo, currTime);
							}
						}
					}
				}
				
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					break;
				}
			}
		}
	}

	public boolean isKeepGoing() {
		return keepGoing;
	}

	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}

	@Override
	public void destroy() {
		this.setKeepGoing(false);
	}
}
